package com.niodata.dp.core;

import com.niodata.dp.log.LogUtil;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import org.apache.log4j.Logger;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

public class DpConfiguration {

  private static final Logger logger = Logger.getLogger(DpConfiguration.class);
  private static Properties properties;

  static {
    PathMatchingResourcePatternResolver resourceLoader = new PathMatchingResourcePatternResolver();
    try {
      properties = new Properties();
      Resource[] resources = resourceLoader.getResources("classpath:*.properties");
      for (Resource resource : resources) {
        Properties props = new Properties();
        InputStream input = resource.getInputStream();
        props.load(input);
        input.close();
        properties.putAll(props);
      }
    } catch (IOException e) {
      LogUtil.logStackTrace(logger, e);
    }
  }

  private DpConfiguration() {
  }

  public static String getString(String key) {
    if (!properties.containsKey(key)) {
      throw new RuntimeException(String.format("property key not found: %s", key));
    }
    String value = properties.getProperty(key);
    return value;
  }

  public static String getString(String key, String defaultValue) {
    String value = properties.getProperty(key);
    if (value == null) {
      logger.warn(String
            .format("Can't find config for key '%s', use default value: %s", key, defaultValue));
      value = defaultValue;
    }
    return value;
  }


  public static List<String> getStringList(String key) {
    if (!properties.containsKey(key)) {
      throw new RuntimeException(String.format("property key not found: %s", key));
    }
    String value = properties.getProperty(key);
    return Arrays.asList(value.split(","));
  }

  public static List<String> getStringList(String key, String sep) {
    String value = getString(key);
    return Arrays.asList(value.split(sep));
  }


  public static int getInt(String key) {
    return Integer.parseInt(properties.getProperty(key));
  }

  public static int getInt(String key, int defaultValue) {
    if (properties.containsKey(key)) {
      return Integer.parseInt(properties.getProperty(key));
    }
    return defaultValue;
  }

  public static boolean getBoolean(String key) {
    return Boolean.parseBoolean(properties.getProperty(key));
  }

  public static boolean getBoolean(String key, boolean defaultValue) {
    if (properties.containsKey(key)) {
      return Boolean.parseBoolean(properties.getProperty(key));
    }
    return defaultValue;
  }

  public static long getLong(String key) {
    return Long.parseLong(properties.getProperty(key));
  }

  public static long getLong(String key, long defaultValue) {
    if (properties.containsKey(key)) {
      return Long.parseLong(properties.getProperty(key));
    }
    return defaultValue;
  }

  public static String getDpHdfsRootDir() {
    return properties.get(ConfigurationKeys.DP_HDFS_ROOT_DIR) == null
          ? "/dp"
          : properties.getProperty("dp.hdfs.root.dir");
  }

  public static String getDpHdfsUsersRootDir() {
    return getDpHdfsRootDir() + "/users";
  }

  public static String getDpHdfsGroupsRootDir() {
    return getDpHdfsRootDir() + "/groups";
  }

  public static String getDpHdfsDataRootDir() {
    return getDpHdfsRootDir() + "/data";
  }

  public static String getAppName() {
    return properties.getProperty("app.name");
  }

  public static String getAppId() {
    return properties.getProperty("app.id");
  }

  public static boolean isHadoopKerberosEnabled() {
    return ConfigurationKeys.HADOOP_SECURITY_KERBEROS
          .equals(properties.getProperty(ConfigurationKeys.HADOOP_SECURITY_TYPE));
  }

  public static boolean isKerberosEnabled() {
    return getBoolean(ConfigurationKeys.KERBEROS_ENABLED, false);
  }

  public static boolean isHadoopHaEnabled() {
    return Boolean.parseBoolean(properties.getProperty(ConfigurationKeys.HADOOP_HA_ENABLED));
  }

  public static boolean isHadoopEnabled() {
    return Boolean.parseBoolean(properties.getProperty(ConfigurationKeys.HADOOP_ENABLED,
          "true"));
  }

  public static Map<String, String> getConfByPrefix(String prefix) {
    return getConfByPrefix(prefix, true);
  }

  public static Map<String, String> getConfByPrefix(String prefix, boolean trimPrefix) {
    Set<Object> keys = properties.keySet();
    Map<String, String> map = new HashMap<>();
    for (Object key : keys) {
      String ks = key.toString();
      String[] segs = ks.split("\\.");
      if (segs[0].equals(prefix) && segs.length > 1) {
        if (trimPrefix) {
          map.put(ks.replace(prefix + ".", ""), properties.getProperty(ks));
        } else {
          map.put(ks, properties.getProperty(ks));
        }
      }
    }
    return map;
  }

}
